package MyOutputFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {
    FSDataOutputStream itstarOut=null;
    FSDataOutputStream otherOut=null;

    public FilterRecordWriter(TaskAttemptContext job) throws IOException {
        Configuration conf = job.getConfiguration();
        //设置获取文件系统
        FileSystem fs = FileSystem.get(conf);
        //创建文件输出流
          //存放包含itstart的网址
        Path itstartPath = new Path("E:\\test\\temp\\itstar.log");
          //存放不包含itstart的网址
        Path otherPath = new Path("E:\\test\\temp\\other.log");
         itstarOut = fs.create(itstartPath);
         otherOut = fs.create(otherPath);

    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
       //写出
        if(key.toString().contains("itstar")){
            itstarOut.write(key.toString().getBytes());

        }else {
            otherOut.write(key.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
       if(itstarOut!=null){
            itstarOut.close();
       }
       if(otherOut!=null){
           otherOut.close();
       }
    }
}
